package com.atguigu.flink.sql.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2023/4/11
 *
 *  flink中提供的window窗口: group window
 *      Table table = input
 *   .window([GroupWindow w].as("w"))  // define window with alias w
 *   .groupBy($("w"))  // group the table by window w
 *   .select($("b").sum());  // aggregate
 *
 *  sql中通用的over窗口
 *
 *  https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/tableapi/#group-windows
 *
 */
public class Demo3_GroupWindowTableAPI
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs());


        //自带水印，自带eventtime
        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());
            //产生了水印，向下游发送水印
            //.assignTimestampsAndWatermarks(watermarkStrategy);

        Schema schema = Schema.newBuilder()
                             //声明普通列
                             .column("id", "STRING")
                             .column("ts", "BIGINT")
                             .column("vc", "INT")
                             //这个列由一个表达式计算得到
                             .columnByExpression("pt", "proctime()")
                             .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                              //写水印怎么来，基于事件时间属性计算得到 沿用流中的水印
                              //.watermark("et","source_watermark()")
                              //自己再生成水印
                              .watermark("et","et - INTERVAL '0.001' SECOND")
                             .build();
        //从流中获取时间属性
        Table table = tableEnvironment.fromDataStream(ds,schema);

        /*
            如果当前事件时间列 后续有 ROWTIME，才说明这里把这个字段真正作为了事件时间属性
         */
        table.printSchema();


        /*
              计数窗口:  需要在 xxx.on(处理时间)
              时间窗口
                    处理时间
                    事件时间

              滚动
              滑动
              会话

         */
        //滚动
        TumbleWithSizeOnTimeWithAlias w1 = Tumble.over(rowInterval(3l)).on($("pt")).as("w");
        TumbleWithSizeOnTimeWithAlias w2 = Tumble.over(lit(5).seconds()).on($("pt")).as("w");
        TumbleWithSizeOnTimeWithAlias w3 = Tumble.over(lit(5).seconds()).on($("et")).as("w");

        //滑动  over(size)  every(slide)   sql，第一次计算必须满足size才会触发！
        SlideWithSizeAndSlideOnTimeWithAlias w4 = Slide.over(rowInterval(3l)).every(rowInterval(2l)).on($("pt")).as("w");
        SlideWithSizeAndSlideOnTimeWithAlias w5 = Slide.over(lit(5).seconds()).every(lit(3).seconds()).on($("pt")).as("w");
        SlideWithSizeAndSlideOnTimeWithAlias w6 = Slide.over(lit(5).seconds()).every(lit(3).seconds()).on($("et")).as("w");


        //会话
        SessionWithGapOnTimeWithAlias w7 = Session.withGap(lit(5).seconds()).on($("pt")).as("w");
        SessionWithGapOnTimeWithAlias w8 = Session.withGap(lit(3).seconds()).on($("et")).as("w");


        //计算
        table
            .window(w8)
            //全局窗口
            //.groupBy($("w"))
            //keyed窗口
            .groupBy($("w"),$("id"))
            //.select($("id"),$("vc").sum().as("sumVc"))
            //时间窗口可以获取时间范围
            .select($("id"),
                $("w").start().as("wstart"),
                $("w").end().as("wend"),
                $("vc").sum().as("sumVc"))
            .execute().print();

        env.execute();


    }
}
